package realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.MyKafkaUtil;

public class Dwd02_CommentInfo {
    public static void main(String[] args) {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkcdc/220926");

        //设置HDFS用户信息
        //System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 2.使用FlinkSQL方式读取Kafka topic_db 主题数据创建表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("comment_info_edu"));

        //打印测试
//        tableEnv.sqlQuery("select * from topic_db").execute().print();

        //TODO 3.过滤出章节评价表数据
        Table commentInfoTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['chapter_id'] chapter_id,\n" +
                "    `data`['course_id'] course_id,\n" +
                "    `data`['comment_txt'] comment_txt,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `data`['deleted'] deleted\n" +
                "from topic_db\n" +
                "where `database` = 'gmall'\n" +
                "and `table` = 'comment_info'\n" +
                "and `type` = 'insert'");

        tableEnv.createTemporaryView("comment_info", commentInfoTable);

        //打印测试
//        tableEnv.sqlQuery("select * from comment_info").execute().print();

        //TODO 4.过滤出章节表数据
        Table chapterInfoTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['chapter_name'] chapter_name,\n" +
                "    `data`['course_id'] course_id,\n" +
                "    `data`['video_id'] video_id,\n" +
                "    `data`['publisher_id'] publisher_id,\n" +
                "    `data`['is_free'] is_free,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `data`['deleted'] deleted\n" +
                "from topic_db\n" +
                "where `database` = 'gmall'\n" +
                "and `table` = 'chapter_info'\n" +
                "and `type` = 'insert'");

        tableEnv.createTemporaryView("chapter_info", chapterInfoTable);

        //打印测试
//        tableEnv.sqlQuery("select * from chapter_info").execute().print();

        //TODO 5.过滤出视频表数据
        Table videoInfoTable = tableEnv.sqlQuery("" +
                "select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['video_name'] video_name,\n" +
                "    `data`['during_sec'] during_sec,\n" +
                "    `data`['video_status'] video_status,\n" +
                "    `data`['video_size'] video_size,\n" +
                "    `data`['chapter_id'] chapter_id,\n" +
                "    `data`['course_id'] course_id,\n" +
                "    `data`['publisher_id'] publisher_id,\n" +
                "    `data`['create_time'] create_time\n" +
                "from topic_db\n" +
                "where `database` = 'gmall'\n" +
                "and `table` = 'video_info'\n" +
                "and `type` = 'insert'");

        tableEnv.createTemporaryView("video_info", videoInfoTable);

        //打印测试
//        tableEnv.sqlQuery("select * from video_info").execute().print();

        //TODO 6.三表关联
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "    ci.id,\n" +
                "    ci.user_id,\n" +
                "    ci.chapter_id,\n" +
                "    ci.course_id course_id_ci,\n" +
                "    ci.comment_txt,\n" +
                "    ci.create_time create_time_ci,\n" +
                "    chi.chapter_name,\n" +
                "    chi.course_id course_id_chi,\n" +
                "    chi.video_id,\n" +
                "    chi.publisher_id,\n" +
                "    chi.is_free,\n" +
                "    chi.create_time create_time_chi,\n" +
                "    vi.video_name,\n" +
                "    vi.during_sec,\n" +
                "    vi.video_status,\n" +
                "    vi.video_size,\n" +
                "    vi.create_time create_time_vi\n" +
                "from comment_info ci\n" +
                "left join chapter_info chi\n" +
                "on ci.chapter_id = chi.id\n" +
                "left join video_info vi\n" +
                "on chi.video_id = vi.id");

        tableEnv.createTemporaryView("result_table",resultTable);

        //TODO 7.创建Kafka DWD层下单主题
        tableEnv.executeSql("create table dwd_comment_info_edu (\n" +
                "    `id` STRING," +
                "    `user_id` STRING,\n" +
                "    `chapter_id` STRING,\n" +
                "    `course_id_ci` STRING,\n" +
                "    `comment_txt` STRING,\n" +
                "    `create_time_ci` STRING,\n" +
                "    `chapter_name` STRING,\n" +
                "    `course_id_chi` STRING,\n" +
                "    `video_id` STRING,\n" +
                "    `publisher_id` STRING,\n" +
                "    `is_free` STRING,\n" +
                "    `create_time_chi` STRING,\n" +
                "    `video_name` STRING,\n" +
                "    `during_sec` STRING,\n" +
                "    `video_status` STRING,\n" +
                "    `video_size` STRING,\n" +
                "    `create_time_vi` STRING,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ")"
                + MyKafkaUtil.getUpsertKafkaDDL("dwd_comment_info"));

        //TODO 8.将数据写出
        tableEnv.executeSql("insert into dwd_comment_info_edu select * from result_table");
    }
}
